package org.quxiao.mmrsql.executor.job.builder;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.quxiao.mmrsql.executor.mapper.TableJoinMapper;
import org.quxiao.mmrsql.executor.mapper.TableValueMapper;
import org.quxiao.mmrsql.executor.reducer.TableJoinReducer;
import org.quxiao.mmrsql.executor.reducer.TableValueReducer;
import org.quxiao.mmrsql.executor.test.TableJoinTester;
import org.quxiao.mmrsql.executor.test.TableValueTester;

public class JobBuilder {

	public static Job newTableValueJob(String table, String family, String qualified, String outputPath) throws IOException
	{
		Configuration conf = new Configuration();
		conf.set("tableName", table);
		Job job = new Job(conf, "TableValue");
		job.setJarByClass(JobBuilder.class);
		Scan scan = new Scan();
		scan.addColumn(family.getBytes(), qualified.getBytes());
		TableMapReduceUtil.initTableMapperJob(table, scan, TableValueMapper.class, Text.class, Text.class, job);
		job.setReducerClass(TableValueReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		TextOutputFormat.setOutputPath(job, new Path(outputPath));
		
		return job;
	}
	
	public static Job newTableJoinJob(String table1, String table2, String inputPath1, String inputPath2, String outputPath) throws IOException
	{
		Configuration conf = new Configuration();
		//add names of join tables
		conf.set("table1", table1);
		conf.set("table2", table2);
		Job job = new Job(conf, "Table Join");
		job.setJarByClass(JobBuilder.class);
		FileInputFormat.addInputPath(job, new Path(inputPath1));
		FileInputFormat.addInputPath(job, new Path(inputPath2));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
		job.setMapperClass(TableJoinMapper.class);
		job.setReducerClass(TableJoinReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		return job;
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

}
